package org.zjt.spark.sql

import java.io.File

import org.apache.spark.SparkConf
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}


/**
  * DESC    将数据保存到表中。 动态处理hdfs的文件到spark sql中去处理。
  *   1、关联json文件
  *   2、将json文件转化为数据库表
  *   3、查询出数据库表的数据，保存到新建的表中。
  *
  * @author
  * @create 2017-05-11 上午11:19
  **/
object SQLRDDTest extends App{

  val conf = new SparkConf().setAppName("WordCount").setMaster("local")
  val sqlContext = new SparkSession.Builder().config(conf).getOrCreate()
  val peopleRDD = sqlContext.sparkContext.textFile("./person.json")

  // TODO: 建表语句       1、新建字段  2、建立schema
  val schemaString = "name age"
  val fields = schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, nullable = true))
  val schema = StructType(fields)

  // TODO: 将json数据关联成表，并将表中的数据转为row记录
  sqlContext.read.json("./person.json").createOrReplaceTempView("people")
  val rowRDD = sqlContext.sql("SELECT name,age FROM people where age = 94 and name = 'li'").rdd.map(row => Row(row.get(0), row.get(1).toString))

  // TODO: 将表中的数据保存到schema中。  DataFrame
  val peopleDF = sqlContext.createDataFrame(rowRDD, schema)
  peopleDF.createOrReplaceTempView("people2")


  val results = sqlContext.sql("SELECT name FROM people2")
  import sqlContext.implicits._
  results.map(attributes => "Name: " + attributes(0)).show()


  // TODO:  parquet对文件的拆分、写出和读取（parquet保存表的数据库表的信息，很适合保存spark sql表）
  var tmp = new File("people.parquet").getCanonicalPath
  peopleDF.write.partitionBy("id").parquet(tmp)
  val df = sqlContext.read.parquet(tmp)
  println(df.collect().mkString(","))



  // TODO: 保存查询结果和过程
  results.write.save("./person_result.txt")

}
